
import java.text.SimpleDateFormat

import scala.collection.mutable.Map
import org.apache.spark.sql.{DataFrame, Dataset, SparkSession}

import scala.collection.mutable

/**
 * 车辆轨迹碰撞IMSI
 */

case class KakouData(cphm: String, decv: String, tme: String, start: Long, end: Long)

case class GsmData(imsi: String, ip: String, tme: String, tmeLong: Long, decv: String)

case class DevidView(virtualId: String, tableName: String, devId: String)

object DiffDataMatch {

  // 轨迹周期内筛选最低个数
  val numberOfTrack = 12

  // 碰撞结果取前几名
  val collectionRowNum = 5

  // viewId对应 devid_view_manager 里面热点卡口对应的 viewid
  val viewId = 20210419

  val devMap: mutable.Map[String, String] = mutable.Map()

  val spark: SparkSession = SparkSession.builder().master("local[*]").appName("DiffDataMatch").getOrCreate()

  import spark.implicits._

  // 数据库连接参数
  val url = "jdbc:oracle:thin:@192.168.0.228:1521/ORCL"
  val jdbcProp = new java.util.Properties
  jdbcProp.setProperty("driver", "oracle.jdbc.driver.OracleDriver")
  jdbcProp.setProperty("user", "dsj")
  jdbcProp.setProperty("password", "dsj")


  def main(args: Array[String]): Unit = {

    /*加载原始数据*/

    // 查询卡口数据
    //val kakouQueryCondition = Array("rownum < 100000")
    val kakouQueryCondition = Array("cphm='豫A05667D' and tme >= to_date( '2021/04/13', 'yyyy/MM/dd' ) and tme < to_date( '2021/04/20', 'yyyy/MM/dd' ) ")
    //    val kakouQueryCondition = Array(" tme >= to_date( '2021/04/13', 'yyyy/MM/dd' ) and tme < to_date( '2021/04/20', 'yyyy/MM/dd' ) ")
    val kakouData = spark.read.jdbc(url, "KAKOUDATA", kakouQueryCondition, jdbcProp)
      .map(kakou => KakouData(kakou(0).toString, kakou(1).toString, kakou(3).toString, getLongTime(kakou(3).toString, -250), getLongTime(kakou(3).toString, 250)))
      .rdd
      .distinct()
      .toDF()

    // 查询设备管理信息表数据并封装成Map
    val devQueryCondition = Array("viewid='" + viewId + "'")
    val devidView = spark.read.jdbc(url, "DEVID_VIEW", devQueryCondition, jdbcProp).map(dev => DevidView(dev(1).toString, dev(4).toString, dev(5).toString)).toDF()
    devidView.foreach(v => {
      val virtualId = v.getString(0)
      val tableName = v.getString(1)
      val devId = v.getString(2)
      if (tableName.toString.equals("GSMDATA")) {
        devMap.+=(devId.toString -> virtualId.toString)
        println()
      } else if (tableName.toString.equals("KAKOUDATA")) {
        devMap.+=(virtualId.toString -> devId.toString)
        println()
      }
    })

    // 查询热点数据（并绑定其卡口设备ID）
    val gsmQueryCondition = Array("ip='XLZZS555' and tme >= to_date( '2021/04/13', 'yyyy/MM/dd' )  and tme < to_date( '2021/04/20', 'yyyy/MM/dd' )")
    val gsmData = spark.read.jdbc(url, "GSMDATA", gsmQueryCondition, jdbcProp)
      .map(gsm => GsmData(gsm(0).toString, gsm(3).toString, gsm(2).toString, getLongTime(gsm(2).toString, 0), getGsmDecvByIp(devMap, gsm(3).toString)))
      .toDF()

    // 注册dev
    //devidView.createOrReplaceTempView("dev")

    // 注册
    // kakouData.createOrReplaceTempView("kakou")

    gsmData.createOrReplaceTempView("gsm")

    // 卡口数据处理
    val kakouCphmCount = kakouData.groupBy("cphm").count()

    // 管理筛选符合最小轨迹数量的轨迹信息
    val kakouDataWithRangeTime = kakouData.join(kakouCphmCount, "cphm")
      .toDF("cphm", "decv", "tme", "start", "end", "count")
      .filter(_.getAs[Long]("count") > numberOfTrack)
      .select("cphm", "decv", "start", "end")

    kakouDataWithRangeTime.createOrReplaceTempView("ka")

    // 获取时间区间内的热点数据
    val frame = spark.sql("select ka.cphm, gsm.imsi from ka join gsm on ka.decv = gsm.decv and (gsm.tmeLong>= ka.start and gsm.tmeLong < ka.end)")

    // 分组统计热点数据
    val imsiCount = frame.groupBy("cphm", "imsi").count().createOrReplaceTempView("imsiCount")

    // 获取每个cphm与imsi碰撞次数较多的数据
    val sql = "select * from (select cphm, imsi ,count, row_number() over ( partition by cphm order by count desc) rank from imsiCount) e where e.rank <= " + collectionRowNum
    spark.sql(sql).show()

  }


  /**
   * 获取给定时间的Long值
   *
   * @param dateStr      给定的时间字符串 格式为 yyyy-MM-dd HH:mm:ss
   * @param bufferSecond 转后Long值后增加的秒数据
   * @return 时间的毫秒值
   */
  def getLongTime(dateStr: String, bufferSecond: Int): Long = {
    val date = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").parse(dateStr)
    date.getTime + (bufferSecond * 1000)
  }


  /**
   * 根据GSM的IP数据获取绑定的Kakou设备
   *
   * @param gsmIp  GSM的IP
   * @param devMap 卡口设备与热点设备映射对象
   * @return 绑定的Kakou设备ID
   */
  def getGsmDecvByIp(devMap: mutable.Map[String, String], gsmIp: String): String = {
    devMap(devMap(gsmIp))
  }

}
